package com.hyw.cm.session.pubsub;

import java.util.Objects;

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.ReflectionUtils;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hyw.cm.server.ConnectionContext;

import io.netty.channel.Channel;

public class RedisSessionPubSub implements IPubSub {

  private static final ObjectMapper json = new ObjectMapper();

  private final RedisTemplate<String, String> redisTemplate;

  private final String topic;

  private final String serverId;

  public RedisSessionPubSub(String serverId, RedisTemplate<String, String> redisTemplate, String topic) {
    super();
    this.serverId = serverId;
    this.redisTemplate = redisTemplate;
    this.topic = topic;
  }

  @Override
  public void onMessage(String event) {
    try {
      SessionBindingEvent evt = json.readValue(event, SessionBindingEvent.class);
      if (!Objects.equals(evt.getServerId(), this.serverId)) {
        Channel ch = ConnectionContext.get(evt.getEquiementId());
        if (Objects.nonNull(ch)) {
          ch.close();
        }
      }
    } catch (JsonProcessingException e) {
      ReflectionUtils.rethrowRuntimeException(e);
    }
  }

  @Override
  public String getTopic() {
    return this.topic;
  }

  @Override
  public void send(Object message) {
    SessionBindingEvent event = SessionBindingEvent.class.cast(message);
    try {
      redisTemplate.convertAndSend(topic, json.writeValueAsString(event));
    } catch (JsonProcessingException e) {
      ReflectionUtils.rethrowRuntimeException(e);
    }
  }

}
